package com.wzk.canalsync.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.wzk.canalsync.common.OperateConstant;
import com.wzk.canalsync.queue.OperateNode;
import com.wzk.canalsync.queue.QueueHandler;
import com.wzk.canalsync.task.ClearLogsFileTask;
import com.wzk.canalsync.template.PostBackDataTemplate;
import com.wzk.canalsync.utils.FileUtils;
import com.wzk.canalsync.utils.LogToFile;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.io.*;
import java.math.BigDecimal;
import java.net.URLEncoder;
import java.nio.file.*;
import java.nio.file.attribute.BasicFileAttributes;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.logging.Logger;

/**
 * @Author: WangZiKang
 * @Date: 2024/12/28 22:42
 */
//canal客户端
@Component
@RequiredArgsConstructor
@Slf4j
public class CanalClient implements ApplicationRunner {

    //映射表
    private Map<String,String> mappingMap = new HashMap<>();

    public static final Logger logger = Logger.getLogger(CanalClient.class.getName());

    private final CanalConnector canalConnector;

    private final QueueHandler queueHandler;

    private final ObjectMapper objectMapper;

    @Value("${canal.batchSize}")
    private Integer batchSize;

    @Value("${canal.subscribeDB}")
    private String subscribeDB;

    @Value("${canal.subscribeTable}")
    private String subscribeTable;

    @Value("${canal.threadSleepTime}")
    private Long threadSleepTime;

    @Value("${canal.mappingPath}")
    private String mappingPath;

//    @Value("${canal.templateClassName}")
//    private String templateClassName;

    @PostConstruct
    public void parseMappingJson(){
        try {
            Path mappingJsonPath = Paths.get(mappingPath);
            // 判断路径是否存在
            if (Files.exists(mappingJsonPath)) {
                log.info("开始解析{}中的映射表....",mappingPath);
                // 读取文件中的 JSON 内容
                String jsonContent = new String(Files.readAllBytes(mappingJsonPath));
                // 使用 Jackson ObjectMapper 解析 JSON 字符串为 HashMap
                ObjectMapper objectMapper = new ObjectMapper();
                HashMap<String, String> map = objectMapper.readValue(jsonContent, HashMap.class);
                this.mappingMap = map;
            } else {
                throw new FileNotFoundException("文件不存在！");
            }
            // 判断queue文件是否存在 存放的是上次程序关闭时未处理完毕的数据
            Path path = Paths.get("queue/node_data.txt");
            //文件存在才读取
            if (Files.exists(path)) {
                log.info("开始加载上次队列中未处理完毕的Node数据....");
                readQueueData();
                //清除加载完毕的文件
                FileUtils.deleteDirectoryContents(path);
                log.info("数据加载完毕");
            }
        }catch (Exception e){
            log.error("出现异常：{}",e.getMessage());
            LogToFile.logToFile("parseMappingJson解析映射表出现异常！异常："+e.getMessage(),logger,"parse_mapping_err");
        }
    }

    public void readQueueData(){
        //1.读取文件中的每一行数据
        BufferedReader reader = null;
        try {
            // 创建 BufferedReader 对象来读取文件
            reader = new BufferedReader(new FileReader("queue/node_data.txt"));
            String line;
            // 使用 readLine() 方法逐行读取文件
            while ((line = reader.readLine()) != null) {
                OperateNode operateNode = objectMapper.readValue(line, OperateNode.class);
                //插入回队列
                queueHandler.push(operateNode);
            }
        } catch (IOException e) {
            log.error("readQueueData方法中出现异常:{}",e.getMessage());
            LogToFile.logToFile("加载queue/node_data.txt中的node数据出现异常！异常:"+e.getMessage(),logger,"read_node_data_err");
        } finally {
            try {
                // 关闭 BufferedReader，释放资源
                if (reader != null) {
                    reader.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    //在springboot启动时会自动执行这方法
    @Override
    public void run(ApplicationArguments args) throws Exception {
        if (canalConnector != null){
            log.info("成功连接到canal-deployer,开始监听binlog!当前监听的数据库：【{}】,当前监听的数据表：【{}】,等待接收数据......",subscribeDB,subscribeTable);
            log.info("映射表：{}",mappingMap);
        }
        try{
            //打开连接
            canalConnector.connect();
            //订阅数据库表
            if (StringUtils.isNotBlank(subscribeDB) && StringUtils.isNotBlank(subscribeTable)){
                canalConnector.subscribe(subscribeDB+"."+subscribeTable);
            }else {
                //所有
                canalConnector.subscribe(".*\\..*");
            }
            //回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
            canalConnector.rollback();
            while (true) {
                // 获取指定数量的数据
                Message message = canalConnector.getWithoutAck(batchSize);
                //获取批量ID
                long batchId = message.getId();
                //获取批量的数量
                int size = message.getEntries().size();
                //如果没有数据
                if (batchId == -1 || size == 0) {
                    try {
                        //线程休眠
                        Thread.sleep(threadSleepTime);
                    } catch (InterruptedException e) {
                        //打印异常信息
                        log.error("CanalClient线程休眠出现异常！exceptionMessage:【{}】",e.getMessage());
                        //写到日志文件
                        LogToFile.logToFile("CanalClient线程休眠出现异常！exceptionMessage:【"+e.getMessage()+"】",logger,"canal_client_sleep_err");
                    }
                } else {
                    //如果有数据,处理数据
                    try {
                        dataHandler(message.getEntries());
                    }catch (Exception e){
                        log.error("出现异常：{}",e.getMessage());
                        LogToFile.logToFile("数据处理出现异常!异常信息:"+e.getMessage(),logger,"data_handler_err");
                    }
                }
                //进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
                canalConnector.ack(batchId);
            }
        }catch (Exception e){
            //出现异常打印异常信息 写到 日志文件
            //打印异常信息
            log.error("CanalClient处理数据出现异常！exceptionMessage:【{}】",e.getMessage());
            //写到日志文件
            LogToFile.logToFile("CanalClient处理数据出现异常！exceptionMessage:【"+e.getMessage()+"】",logger,"canal_client_err");
        }
    }

    /**
     * 打印canal-deployer解析binlog获得的实体类信息
     */
    private void dataHandler(List<CanalEntry.Entry> entrys) throws Exception {
        //开始处理数据
        for (CanalEntry.Entry entry : entrys) {
            LogToFile.logToFile("1.dataHandler开始处理数据！",logger,"canal_data_handler");
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                //开启/关闭事务的实体类型，跳过
                continue;
            }
            //RowChange对象，包含了一行数据变化的所有特征
            //比如isDdl 是否是ddl变更操作 sql 具体的ddl sql beforeColumns afterColumns 变更前后的数据字段等等
            CanalEntry.RowChange rowChage;
            try {
                rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("错误！ 从 eromanga-event 解析出现错误！ , 数据:" + entry.toString(), e);
            }
            //获取操作类型：insert/update/delete类型
            CanalEntry.EventType eventType = rowChage.getEventType();
            LogToFile.logToFile("2.当前操作的类型:【"+eventType+"】",logger,"canal_data_handler");
            //打印Header信息
            String headerMsg = String.format("================>> binlog[%s:%s] , name[%s,%s] , eventType : %s <<================",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType);
            log.info(headerMsg);
            LogToFile.logToFile("3.headerMessage:【"+headerMsg+"】",logger,"canal_data_handler");
            //判断是否是DDL语句
            if (rowChage.getIsDdl()) {
                System.out.println("================>>;isDdl: true,sql:" + rowChage.getSql() +"<<================");
            }
            //获取RowChange对象里的每一行数据，打印出来
            for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                //如果是删除语句
                if (eventType == CanalEntry.EventType.DELETE) {
                    //获取开始时间
                    long startTime = System.currentTimeMillis();
                    for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
                        if (column.getIsKey()){
                            //1.获取到要删除的数据的id值
                            String id = column.getValue();
                            log.info("-------------解析出来要删除的数据id:{}-----------",id);
                            LogToFile.logToFile("4.解析出来的！删除的数据id:"+id,logger,"canal_data_handler");
                            //2.封装操作
                            OperateNode node = new OperateNode();
                            node.setOperate(OperateConstant.DELETE);
                            HashMap<String, Object> map = new HashMap<>();
                            map.put("id", id);
                            LogToFile.logToFile("5.operateNode数据:"+node,logger,"canal_data_handler");
                            node.setParamsMap(map);
                            queueHandler.push(node);
                            //结束时间
                            long endTime = System.currentTimeMillis();
                            log.info("---------耗时【"+(endTime-startTime)+"】ms----------");
                            LogToFile.logToFile("5.---------耗时【"+(endTime-startTime)+"】ms----------",logger,"canal_data_handler");
                            break;
                        }
                    }
                } else if (eventType == CanalEntry.EventType.INSERT) { // 如果是新增语句
                    //获取开始时间
                    long startTime = System.currentTimeMillis();
                    //1.存储提取出来的数据
                    HashMap<String, Object> insertMap = new HashMap<>();
                    //2.遍历每一列
                    for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
                        //3.拿到每一列的数据名到映射表中查找出对应的es的字段名
                        String esFiledName = mappingMap.get(column.getName());
                        if (StringUtils.isBlank(esFiledName)) {
                            continue;
                        }
                        //4.判断该字段是否为空
                        if (!column.getIsNull()){
//                            Class<?> clazz = getTemplateClass(templateClassName);
                            Class<?> clazz = PostBackDataTemplate.class;
                            //5.获取该属性在java中对应的类型
                            Class<?> type = clazz.getDeclaredField(esFiledName).getType();
                            //6.转换成对应的类型并返回
                            Object javaTypeFiled = convertStringToFieldType(column.getValue(), type);
                            //7.插入到map中
                            insertMap.put(esFiledName, javaTypeFiled);
                        }else {
                            //是null直接插入null
                            insertMap.put(esFiledName,null);
                        }
                    }
                    LogToFile.logToFile("4.解析出来的！插入的数据:"+insertMap,logger,"canal_data_handler");
                    //获取主键
                    OperateNode node = new OperateNode();
                    node.setOperate(OperateConstant.INSERT);
                    node.setParamsMap(insertMap);
                    queueHandler.push(node);
                    LogToFile.logToFile("5.operateNode数据:"+node,logger,"canal_data_handler");
                    //结束时间
                    long endTime = System.currentTimeMillis();
                    LogToFile.logToFile("5.---------耗时【"+(endTime-startTime)+"】ms----------",logger,"canal_data_handler");
                } else {//如果是更新的语句
                    //获取开始时间
                    long startTime = System.currentTimeMillis();
                    //1.存储提取出来的数据
                    HashMap<String, Object> updateMap = new HashMap<>();
                    //2.找到要修改的值
                    for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
                        //3.拿到每一列的数据名到映射表中查找出对应的es的字段名
                        String esFiledName = mappingMap.get(column.getName());
                        //判断是否为主键
                        if (column.getIsKey()){
                            updateMap.put("id", column.getValue());
                            continue;
                        }
                        //判断是否为空
                        if (StringUtils.isBlank(esFiledName)) {
                            continue;
                        }
                        //4.判断是否需要修改
                        if (!column.getUpdated()){
                            continue;
                        }
                        //5.走到这 说明是需要修改的字段 判断是否为null
                        if (!column.getIsNull()){
                            //6.获取该属性在java中对应的类型
//                            Class<?> clazz = getTemplateClass(templateClassName);
                            Class<?> clazz = PostBackDataTemplate.class;
                            Class<?> type = clazz.getDeclaredField(esFiledName).getType();
                            //7.转换成对应的类型并返回
                            Object javaTypeFiled = convertStringToFieldType(column.getValue(), type);
                            //8.插入到map中
                            updateMap.put(esFiledName, javaTypeFiled);
                        }else {
                            //9.是null直接插入null
                            updateMap.put(esFiledName,null);
                        }
                    }
                    LogToFile.logToFile("4.解析出来的！修改的内容:"+updateMap,logger,"canal_data_handler");
                    //修改数据
                    //异步执行插入操作
                    OperateNode node = new OperateNode();
                    node.setOperate(OperateConstant.UPDATE);
                    node.setParamsMap(updateMap);
                    queueHandler.push(node);
                    LogToFile.logToFile("5.operateNode数据:"+node,logger,"canal_data_handler");
                    //结束时间
                    long endTime = System.currentTimeMillis();
                    LogToFile.logToFile("5.---------耗时【"+(endTime-startTime)+"】ms----------",logger,"canal_data_handler");
                }
            }
        }
    }

    //获取mysql类型对应的java模板中属性的字段类型 完成转换
    public <T> T convertStringToFieldType(String value, Class<T> filedJavaType) throws Exception {
        if (filedJavaType == String.class) {
            return filedJavaType.cast(value); // 如果字段类型是 String，直接返回
        } else if (filedJavaType == Integer.class || filedJavaType == int.class) {
            return filedJavaType.cast(Integer.valueOf(value)); // 转换为 Integer 或 int
        } else if (filedJavaType == Long.class || filedJavaType == long.class) {
            return filedJavaType.cast(Long.valueOf(value)); // 转换为 Long 或 long
        } else if (filedJavaType == Double.class || filedJavaType == double.class) {
            return filedJavaType.cast(Double.valueOf(value)); // 转换为 Double 或 double
        } else if (filedJavaType == Float.class || filedJavaType == float.class) {
            return filedJavaType.cast(Float.valueOf(value)); // 转换为 Float 或 float
        } else if (filedJavaType == Boolean.class || filedJavaType == boolean.class) {
            return filedJavaType.cast(Boolean.valueOf(value)); // 转换为 Boolean 或 boolean
        } else if (filedJavaType == LocalDateTime.class) {
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            return filedJavaType.cast(LocalDateTime.parse(value, formatter)); // 使用 LocalDateTime.parse() 来转换
        } else if (filedJavaType == LocalDate.class) {
            return filedJavaType.cast(LocalDate.parse(value)); // 转换为 LocalDate
        } else if (filedJavaType == BigDecimal.class) {
            return filedJavaType.cast(new BigDecimal(value)); // 转换为 BigDecimal
        } else {
            throw new IllegalArgumentException("当前类型无法转换! type: " + filedJavaType.getName());
        }
    }
}
